package com.yequan.mq.consumer;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.util.List;

/**
 * @author yq
 * @date 2020/09/21
 */
public class ConsumerRetry {

    public static void main(String[] args) throws InterruptedException, MQClientException {
        // 声明一个消费者consumer，需要传入一个组
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumerTest");
        // 设置集群的NameServer地址，多个地址之间以分号分隔
        consumer.setNamesrvAddr("127.0.0.1:9876");
        // 设置consumer的消费策略
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 集群模式消费，广播消费不会重试
        consumer.setMessageModel(MessageModel.CLUSTERING);
        // 设置最大重试次数，默认是16次
        consumer.setMaxReconsumeTimes(16);
        //设置消费超时时长
        consumer.setConsumeTimeout(30);
        // 设置consumer所订阅的Topic和Tag，*代表全部的Tag
        consumer.subscribe("TopicTest", "*");
        // Listener，主要进行消息的逻辑处理,监听topic，如果有消息就会立即去消费
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                // 获取第一条消息，进行处理
                try {
                    MessageExt messageExt = msgs.get(0);
                    String msgBody = new String(messageExt.getBody(), "utf-8");
                    System.out.println(" 接收新的消息:消息内容为：" + msgBody + " == 消息整体为：" + msgs);
                    // 模拟消息消费失败操作
                    if (StringUtils.equals(msgBody, "生产第2条消息")) {
                    }
                    int a = 1 / 0;
                } catch (Exception e) {
                    e.printStackTrace();
                    System.out.println(e);
                    // 尝试重新消费，直接第三次如果还不成功就放弃消费，进行消息消费失败补偿操作
                    if (msgs.get(0).getReconsumeTimes() == 3) {
                        // 返回成功状态码，消息会在mq上被清掉，但是这是一条失败的消息，所以我们需要做失败补偿操作
                        // 补偿策略：记录数据库或者日志，启动一个定时脚本去扫描这些失败的消息，进行相应处理
                        //           或者将失败的消息放入另一个topic，生产者订阅该topic，将失败的消息发送给生产者，生产者重新发送到mq上
                        System.out.println("消息记录日志：" + msgs);
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    } else {
                        // 重试状态码，重试机制可配置
                        System.out.println("消息消费失败，尝试重试！！！");
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 调用start()方法启动consumer
        consumer.start();
        System.out.printf("Consumer1 启动.%n");
    }


}
